package com.k8.common.executor.threadPool;

import com.k8.common.executor.blockingQueue.K8Agent;
import com.k8.common.executor.blockingQueue.MemoryLimitedLinkedBlockingQueue;
import com.k8.common.executor.blockingQueue.safe.MemorySafeLinkedBlockingQueue;
import com.k8.rpc.url.URL;

import java.util.concurrent.*;

import static com.k8.common.constants.CommonConstants.*;
import static com.k8.common.constants.CommonKeyConstants.*;


/**
 * K8Agent由META-INF/MANIFEST.MF文件指定加载，premain方法由jvm在主方法启动前调用，执行时会传递instrumentation
 * 需要通过其来判断该对象加入阻塞队列后是否会超过内存大小限制。
 */
public abstract class AbstractThreadPool implements ThreadPool {
    public static int SYSTEM_CORE = Runtime.getRuntime().availableProcessors();

    protected BlockingQueue<Runnable> getBlockingQueue(URL url) {
        String blockingQueueType = url.getParameter(BLOCKING_QUEUE_TYPE_KEY, LINKED_QUEUE);
        BlockingQueue<Runnable> blockingQueue = null;
        if (LINKED_QUEUE.equals(blockingQueueType)) {
            blockingQueue = new LinkedBlockingQueue<>();
        } else if (MEMORY_SAFE_QUEUE.equals(blockingQueueType)) {
            String maxFreeMemory = url.getParameter(MAX_FREE_MEMORY);
            if (maxFreeMemory != null) {
                blockingQueue = new MemorySafeLinkedBlockingQueue<>(Integer.parseInt(maxFreeMemory));
            } else {
                blockingQueue = new MemorySafeLinkedBlockingQueue<>();
            }
        } else if (MEMORY_LIMITED_QUEUE.equals(blockingQueueType)) {
            //剩余内存限制
            String memoryLimit = url.getParameter(QUEUE_MEMORY_LIMIT);
            if (memoryLimit == null) {
                blockingQueue = new MemoryLimitedLinkedBlockingQueue<>(K8Agent.instrumentation);
            } else {
                blockingQueue = new MemoryLimitedLinkedBlockingQueue<>(Long.parseLong(memoryLimit), K8Agent.instrumentation);
            }
        } else if (ARRAY_QUEUE.equals(blockingQueueType)) {
            String limitLength = url.getParameter(ARRAY_QUEUE_LIMIT_LEN_KEY);
            if (limitLength == null)
                throw new IllegalArgumentException("ArrayBlockingQueue need limit length of queue, which is not exist in url");
            blockingQueue = new ArrayBlockingQueue<>(Integer.parseInt(limitLength));
        } else if (SYNCHRONOUS_QUEUE.equals(blockingQueueType)) {
            blockingQueue = new SynchronousQueue<>();
        } else {
            throw new IllegalArgumentException("Blocking queue type is error:" + blockingQueueType);
        }
        return blockingQueue;
    }

    @Override
    public ExecutorService getExecutor(URL url) {
        int maxPar = url.getParameter(MAX_THREAD_COUNT_KEY, 0);
        int max = maxPar <= 0 ? SYSTEM_CORE * 2 + 1 : maxPar;
        int corePar = url.getParameter(CORE_THREAD_COUNT_KEY, 0);
        int core = corePar <= 0 ? SYSTEM_CORE * 2 + 1 : corePar;
        int limitPar = url.getParameter(Limit_THREADS_COUNT_KEY, 0);
        int limit = limitPar <= 0 ? SYSTEM_CORE * 2 + 1 : limitPar;
        return doGetExecutor(url, core, max, limit);
    }

    abstract protected ExecutorService doGetExecutor(URL url, int core, int max, int limit);
}
